package com.ruoyi.system.flink;

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

@FunctionalInterface
public interface FutureLeader {

    Logger log = LoggerFactory.getLogger(FutureLeader.class);

    List<Supplier<Map<String,Object>>> tasks(final Object evnet);

    default Map<String,Object> load(final Object event,final long timeout){
        long startTime = System.currentTimeMillis();
        List<Supplier<Map<String, Object>>> tasks = this.tasks(event);
        if(tasks==null||tasks.isEmpty())return null;
        List<CompletableFuture<Map<String,Object>>> futures = new ArrayList<>(5);
        tasks.forEach(task->futures.add(CompletableFuture.supplyAsync(task)));
        log.info("任务数：{}",futures.size());
        final HashMap<String,Object> results = new HashMap<>(100);
        long endTime = System.currentTimeMillis() + timeout;
        int index = 0;
        for (CompletableFuture<Map<String, Object>> future : futures) {
            ++index;
            try {
                results.putAll(future.get(Math.max(0l,endTime-System.currentTimeMillis()), TimeUnit.MILLISECONDS));
            } catch (Exception e) {
                log.warn("failed to load future for "+ index+" th task",e);
            }
        }
        log.info("任务结果：{}，耗时{}s",results,(System.currentTimeMillis()-startTime)/1000);
        return results;
    }

}

class MyLocalSelect implements FutureLeader{

    @Override
    public List<Supplier<Map<String, Object>>> tasks(Object event) {
        List<Supplier<Map<String, Object>>> taskList = new ArrayList<>();
        taskList.add(new MyLocalTask(event));
        taskList.add(()->{
            log.info("{}-查询用户任务。。。",event);
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                log.warn("本地查询用户任务超时了",e);
            }
            return Map.of("本地查询用户任务",event+"用户已查到");
        });
        taskList.add(()->{
            log.info("{}-查询商品任务。。。",event);
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                log.warn("本地查询商品任务超时了",e);
            }
            return Map.of("本地查询商品任务",event+"商品已查到");
        });
        return taskList;
    }
}


@AllArgsConstructor
@Slf4j
class MyLocalTask implements Supplier<Map<String,Object>>{

    Object event;

    @Override
    public Map<String, Object> get() {
        log.info("{}-本地查询订单任务很耗时。。。",event);
        try {
            Thread.sleep(6000);
        } catch (InterruptedException e) {
            log.warn("本地查询订单任务超时了",e);
        }
        return Map.of("本地查询订单任务",event+"订单已查到");
    }
}

class MyRemoteSelect implements FutureLeader{

    @Override
    public List<Supplier<Map<String, Object>>> tasks(Object evnet) {
        List<Supplier<Map<String, Object>>> taskList = new ArrayList<>();
        taskList.add(new MyRemoteTask(evnet));
        taskList.add(()->{
            log.info("{}-查询用户任务。。。",evnet);
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                log.warn("远程查询用户任务超时了",e);
            }
            return Map.of("远程查询用户任务",evnet+"用户已查到");
        });
        taskList.add(()->{
            log.info("{}-查询商品任务。。。",evnet);
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                log.warn("远程查询商品任务超时了",e);
            }
            return Map.of("远程查询商品任务",evnet+"商品已查到");
        });
        return taskList;
    }
}

@AllArgsConstructor
@Slf4j
class MyRemoteTask implements Supplier<Map<String,Object>>{

    Object event;

    @Override
    public Map<String, Object> get() {
        log.info("{}-远程查询订单任务很耗时。。。",event);
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            log.warn("远程查询订单任务超时了",e);
        }
        return Map.of("远程查询订单任务",event+"订单已查到");
    }
}

@Slf4j
class Test{
    public static void main(String[] args) {
        log.info("开始执行查询任务。。。。");
        long startTime = System.currentTimeMillis();
        String where = "主线程带进来的查询条件";
        long timeout = 8000;//能接受的超时时间ms
        Map<String, Object> localResult = new MyLocalSelect().load(where, timeout);
        Map<String, Object> remoteResult = new MyRemoteSelect().load(where,timeout);
        log.info("合并任务结果:\n{};\n{};\n总耗时{}s",localResult,remoteResult,(System.currentTimeMillis()-startTime)/1000);
    }
}
